package cc.shacocloud.redis.study.queue.message.example;

import cc.shacocloud.redis.study.queue.message.MessageConsumerMo;
import cc.shacocloud.redis.study.queue.message.MessageQueueConsumer;
import cc.shacocloud.redis.study.queue.message.MessageQueueProducer;
import cc.shacocloud.redis.study.queue.message.RedisStreamMessageQueue;
import cc.shacocloud.redis.study.util.redis.RedisUtil;
import cn.hutool.core.util.RandomUtil;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class RedisStreamMessageQueueExample {

    public static void main(String[] args) throws InterruptedException {
        RedisUtil.flushDB();

        String queueName = "shacocloud-msg-test";
        String group = "test";

        // 消费者A -------------------
        String consumerA = "消费者A";
        MessageQueueConsumer messageQueueConsumerA = new RedisStreamMessageQueue(queueName, group, consumerA);

        // 添加消费者
        messageQueueConsumerA.addConsumer((messageHandle, consumerMo) -> {
            for (MessageConsumerMo.MessageMo message : consumerMo.getMessages()) {

                String messageId = message.getMessageId();
                messageHandle.ack(messageId);

                log.info("消费者：{} 消费： {} - {}", consumerA, messageId, new String(message.getMessage()));
            }
        });

        // 消费者B -------------------
        String consumerB = "消费者B";
        MessageQueueConsumer messageQueueConsumerB = new RedisStreamMessageQueue(queueName, group, consumerB);

        // 添加消费者
        messageQueueConsumerB.addConsumer((messageHandle, consumerMo) -> {
            for (MessageConsumerMo.MessageMo message : consumerMo.getMessages()) {

                String messageId = message.getMessageId();
                messageHandle.ack(messageId);

                log.info("消费者：{} 消费： {} - {}", consumerB, messageId, new String(message.getMessage()));
            }
        });

        // 生产者 -------------------
        String producer = "生产者";
        MessageQueueProducer messageQueueProducer = new RedisStreamMessageQueue(queueName, group, producer);
        for (int i = 0; i < 10; i++) {
            messageQueueProducer.addMessage(RandomUtil.randomString(10).getBytes());
        }

    }

}
